package com.burges.net.typeInformation

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, _}

/**
  * 基础数据类型
  * 创建人    BurgessLee 
  * 创建时间   2020/1/25 
  * 描述
  */
object BasicType {

	def main(args: Array[String]): Unit = {
		//1.原生数据类型
		// 第一种实现方式，实现BasicTypeInfo数据类型
		// 支持原生的Java原生数据类型或String类型
		val env = StreamExecutionEnvironment.getExecutionEnvironment
		//1.1 创建Int类型的数据集
		val intStream: DataStream[Int] = env.fromElements(3,1,1,2,5)
		//1.2 创建String类型的数据集
		val stringStream: DataStream[String] = env.fromElements("hello", "flink")

		//2.另外一种是实现方式
		//  实现类外一种TypeInformation是BasicArrayTypeInfo
		//  能够支持Java基本类型数组或String对象数组
		//2.1 通过从数组中创建数据集
		val intArrayByArray: DataStream[Int] = env.fromCollection(Array(3,1,2,1,5))
		//2.2 通过List集合创建数据集
		val intArrayByList: DataStream[Int] = env.fromCollection(List(3,1,2,1,5))

		//Java Tuples类型
		//创建实例化Tuple2创建具有两个元素的数据集
		val tupleStreamDemo: DataStream[(String, Int)] = env.fromElements(new Tuple2("a",1),new Tuple2("b",2))

		//Scala Case Class
		//定义WordCount Case Class数据结构
		case class WordCount(word: String, count: Int)
		//通过fromElements方法创建数据集
		val input: DataStream[WordCount] = env.fromElements(WordCount("hello",1), WordCount("world",2))
		val keyStream1 = input.keyBy("word")//根据word字段为分区字段
		val keyStream2: KeyedStream[WordCount, Tuple] = input.keyBy(0)
		//可以通过制定position分区
		//通过Tuple创建数据集
		val tupleStream: DataStream[Tuple2[String, Int]] = env.fromElements(("q",1), ("c",2))
		//使用默认字段名称获取字段，其中_1表示tuple这和第一个字段
		tupleStream.keyBy("_1")


		//POJOs类型
		val personStream = env.fromElements(new Person("Peter", 14), new Person("Linda", 25))
		personStream.keyBy("name")

		//Flink Value类型

		//特殊数据类型
		//创建map类型的数据集
		var mapStream = env.fromElements(Map("name"-> "Peter", "age"-> 18), Map("name"-> "Linda", "age"-> 25))
		val listStream = env.fromElements(List(1, 2, 2, 5), List(2, 5, 6, 7))

	}

	//通过Scala定义类
	class Person(var name: String, var age: Int){
		//默认空构造器
		def this(){
			this(null,-1)
		}
	}

}


